package df

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.spark.HBaseContext
import org.apache.hadoop.hbase.spark.datasources.HBaseTableCatalog
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.when

object E6_HBase {

  // 创建 Spark 运行配置对象
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
  //创建 SparkSession 对象
  val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
  // 导入隐式转换
  import spark.implicits._
  def main(args: Array[String]): Unit = {
    read_hbase
    readerV2
    save_hbase
    saveV2
  }

  def read_hbase = {
    // 创建 HBase 配置对象
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "localhost:2181")
    // 创建 HBaseContext 对象，用于在 Spark 中操作 HBase
    new HBaseContext(spark.sparkContext, conf)

    // 指定要操作的 HBase 表名
    val hbase_table = "people"

    // 定义 HBase 列映射关系
    val hbase_column_mapping = "id STRING :key, " +
      "name STRING cf:name, " +
      "gender STRING cf:gender, " +
      "age STRING cf:age"

    // 使用 Spark 读取 HBase 表中的数据，根据列映射关系转换为 DataFrame
    val hbaseDF = spark.read
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.columns.mapping", hbase_column_mapping)
      .option("hbase.spark.pushdown.columnfilter", false) // 禁用列过滤下推优化
      .option("hbase.table", hbase_table)
      .load()

    // 打印 DataFrame 的模式信息
    hbaseDF.printSchema()

    // 显示DataFrame内容
    hbaseDF.show()

    // 使用DSL语法：统计男性和女性的人数
    val genderCountsDF = hbaseDF.groupBy("gender").count()
    println("性别统计:")
    genderCountsDF.show()

    // 使用SQL语法：查询年龄大于30的人数
    hbaseDF.createOrReplaceTempView("people") // 首先要注册Dataframe为临时视图
    val result2 = spark.sql("SELECT COUNT(*) FROM people WHERE age > 30")
    println("年龄大于30岁的人员:")
    result2.show()

    //关闭 Spark
//    spark.stop()
  }

  def readerV2 = {
    // 创建 HBase 配置对象
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "localhost:2181")
    // 创建 HBaseContext 对象，用于在 Spark 中操作 HBase
    val hbaseContext = new HBaseContext(spark.sparkContext, conf)

    // 定义目录
    val catalog =
      s"""{
         |"table":{"namespace":"default", "name":"people"},
         |"rowkey":"key",
         |"columns":{
         |"id":{"col":"key", "type":"string"},
         |"name":{"cf":"cf", "col":"name", "type":"string"},
         |"gender":{"cf":"cf", "col":"gender", "type":"string"},
         |"age":{"cf":"cf", "col":"age", "type":"string"}
         |}
         |}""".stripMargin

    // 使目录读取 HBase 表中的数据，转换为 DataFrame
    val hbaseDF = spark.read
      .options(Map(HBaseTableCatalog.tableCatalog -> catalog))
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.spark.pushdown.columnfilter", false) // 禁用列过滤下推优化
      .load()

    // 打印 DataFrame 的模式信息
    hbaseDF.printSchema()

    // 显示DataFrame内容
    hbaseDF.show()

    // 使用DSL语法：统计男性和女性的人数
    val genderCountsDF = hbaseDF.groupBy("gender").count()
    println("性别统计:")
    genderCountsDF.show()

    // 使用SQL语法：查询年龄大于30的人数
    hbaseDF.createOrReplaceTempView("people") // 首先要注册Dataframe为临时视图
    val result2 = spark.sql("SELECT COUNT(*) FROM people WHERE age > 30")
    println("年龄大于30岁的人员:")
    result2.show()

    //关闭 Spark
//    spark.stop()
  }

  def save_hbase = {
    // 创建 HBase 配置对象
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "localhost:2181")
    // 创建 HBaseContext 对象，用于在 Spark 中操作 HBase
    new HBaseContext(spark.sparkContext, conf)

    // 指定要操作的 HBase 表名
    val people_table = "people"

    // 定义 HBase 列映射关系
    val people_column_mapping = "id STRING :key, " +
      "name STRING cf:name, " +
      "gender STRING cf:gender, " +
      "age STRING cf:age"

    // 使用 Spark 读取 HBase 表中的数据，根据列映射关系转换为 DataFrame
    val hbaseDF = spark.read
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.columns.mapping", people_column_mapping)
      .option("hbase.spark.pushdown.columnfilter", false) // 禁用列过滤下推优化
      .option("hbase.table", people_table)
      .load()

    // 将性别值改成中文并保存到库中
    val hbaseDFWithChineseGender = hbaseDF.withColumn("gender",
      when($"gender" === "M", "男性").otherwise("女性")
    )
    hbaseDFWithChineseGender.write
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.columns.mapping", people_column_mapping)
      .option("hbase.table", people_table)
      .save()

    // 定义性别统计表的列映射关系
    val gender_count_column_mapping = "gender STRING :key, " +
      "count INT cf:count"

    // 统计男性和女性的人数
    val genderCountsDF = hbaseDFWithChineseGender.groupBy("gender").count()

    // 将性别统计结果存储到另一张表中，使用HBase列映射关系
    genderCountsDF.write
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.columns.mapping", gender_count_column_mapping)
      .option("hbase.table", "gender_count")
      .save()

    //关闭 Spark
    spark.stop()
  }

  def saveV2={
    // 创建 HBase 配置对象
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "localhost:2181")
    // 创建 HBaseContext 对象，用于在 Spark 中操作 HBase
    new HBaseContext(spark.sparkContext, conf)

    // 定义people表的目录
    val people_catalog = s"""{
                            |"table":{"namespace":"default", "name":"people"},
                            |"rowkey":"key",
                            |"columns":{
                            |"id":{"col":"key", "type":"string"},
                            |"name":{"cf":"cf", "col":"name", "type":"string"},
                            |"gender":{"cf":"cf", "col":"gender", "type":"string"},
                            |"age":{"cf":"cf", "col":"age", "type":"string"}
                            |}
                            |}""".stripMargin

    // 使用 Spark 读取 HBase 表中的数据，根据列映射关系转换为 DataFrame
    val hbaseDF = spark.read
      .options(Map(HBaseTableCatalog.tableCatalog -> people_catalog))
      .format("org.apache.hadoop.hbase.spark")
      .option("hbase.spark.pushdown.columnfilter", false) // 禁用列过滤下推优化
      .load()

    // 将性别值改成中文并保存到库中
    val hbaseDFWithChineseGender = hbaseDF.withColumn("gender",
      when($"gender" === "M", "男性").otherwise("女性")
    )
    hbaseDFWithChineseGender.write
      .options(Map(HBaseTableCatalog.tableCatalog -> people_catalog))
      .format("org.apache.hadoop.hbase.spark")
      .save()

    // 定义gender_count表的目录
    val gender_count_catalog = s"""{
                                  |"table":{"namespace":"default", "name":"gender_count"},
                                  |"rowkey":"key",
                                  |"columns":{
                                  |"gender":{"col":"key", "type":"string"},
                                  |"count":{"cf":"cf", "col":"count", "type":"int"}
                                  |}
                                  |}""".stripMargin

    // 统计男性和女性的人数
    val genderCountsDF = hbaseDFWithChineseGender.groupBy("gender").count()
    genderCountsDF.show()

    // 将性别统计结果存储到另一张表中，使用HBase列映射关系
    genderCountsDF.write
      .options(Map(HBaseTableCatalog.tableCatalog -> gender_count_catalog))
      .format("org.apache.hadoop.hbase.spark")
      .save()

    //关闭 Spark
    spark.stop()
  }
}
